package org.ethereum.net.dpos;

import java.util.concurrent.atomic.AtomicInteger;

import org.ethereum.listener.CompositeEthereumListener;
import org.ethereum.net.dpos.EOrgMessage.ResponseCode;
import org.ethereum.net.dpos.ElectionNode.Status;
import org.ethereum.net.rlpx.Node;
import org.ethereum.net.server.Channel;
import org.ethereum.net.server.ChannelManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * 
 * Election Organization protocol handler based on P2P protocol network.
 * 
 * Netty stack:
 * |------------ETH, BZZ, SHH, DPoS, EOrg protocols---------|
 * |------------P2P protocol -------------------------|
 * |------------TCP & UDP ----------------------------|
 * 
 * 选举流程简述: 
 * 目前有5台服务器，每台服务器均没有数据，它们的编号分别是1,2,3,4,5,按编号依次启动，它们的选择举过程如下：
 * 	1.服务器1启动，给自己投票，然后发投票信息，由于其它机器还没有启动所以它收不到反馈信息，服务器1的状态一直属于LOOKING。
 * 	2.服务器2启动，给自己投票，同时与之前启动的服务器1交换结果，由于服务器2的编号大所以服务器2胜出，但此时投票数没有大于半数，所以两个服务器的状态依然是LOOKING。
 * 	3.服务器3启动，给自己投票，同时与之前启动的服务器1,2交换信息，由于服务器3的编号最大所以服务器3胜出，此时投票数正好大于半数，所以服务器3成为领导者，服务器1,2成为小弟。
 * 	4.服务器4启动，给自己投票，同时与之前启动的服务器1,2,3交换信息，尽管服务器4的编号大，但之前服务器3已经胜出，所以服务器4只能成为小弟。
 * 	5.服务器5启动，后面的逻辑同服务器4成为小弟。
 * 
 * @author shaolinwu 03/15/2018
 */
@Component
@Sharable
public class EOrganizationHandler extends SimpleChannelInboundHandler<EOrgMessage>{

	private static final Logger logger = LoggerFactory.getLogger("eorg");
	
	public final static byte VERSION = 0;
	
	@Autowired
	CompositeEthereumListener listener;
	
	@Autowired
	ChannelManager channelManager;
	
	@Autowired
	ElectionNode electionNode;
	
	// my current tickets
	private final AtomicInteger myTickets = new AtomicInteger(1);
	
	public EOrganizationHandler() {
	}
	
	/**
	 * this activate happens for every established channel.
	 */
	public void activate() {
        logger.info("EOrg protocol activated");
        
        if (!electionNode.isLeaderNode()) {
        		// set the initial tickets from peer numbers.
        		myTickets.set(channelManager.getActivePeers().size());
        		if (myTickets.get() == 0) {
        			myTickets.set(1);
        		}
        		EOrgVoteRequest request = new EOrgVoteRequest(myTickets.get(), electionNode.getCurrentNodeId());
        		if (!this.sendVoteRequest(request)) {
        			electionNode.state = Status.LOOKING;
        			logger.info("Election p2p channel is not established! unable to register myself as candicate, try it later.");
        			return;
        		}
        }
    }
	
	public void deactivate() {
		
	}
	
	/**
	 * follow P2P network handler.
	 */
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, EOrgMessage msg) throws Exception {
        switch (msg.getCommand()) {
            case VOTE_REQUEST:
            		EOrgVoteRequest vrequest = (EOrgVoteRequest)msg;
            		if (logger.isDebugEnabled())
            			logger.debug("received vote request: {}", vrequest.toString());
            		
            		if (electionNode.state == Status.SELECTED) {
            			EOrgVoteResponse vresponse = new EOrgVoteResponse(myTickets.get(), vrequest.getElectionId(), ResponseCode.SUCCESS);
                		ctx.writeAndFlush(vresponse);
            		} else if (vrequest.getTickets() == 1 && myTickets.get() == 1) {
            			EOrgVoteResponse vresponse = new EOrgVoteResponse(myTickets.get(), vrequest.getElectionId(), ResponseCode.LOOKING);
                		ctx.writeAndFlush(vresponse);
                		electionNode.state = Status.LOOKING;
            		} else if (vrequest.getTickets() > myTickets.get()) {
            			EOrgVoteResponse vresponse = new EOrgVoteResponse(myTickets.get(), vrequest.getElectionId(), ResponseCode.WIN);
                		ctx.writeAndFlush(vresponse);
                		electionNode.state = Status.SELECTED;
            		} else {
            			myTickets.incrementAndGet();
            			EOrgVoteResponse vresponse = new EOrgVoteResponse(myTickets.get(), vrequest.getElectionId(), ResponseCode.SUCCESS);
                		ctx.writeAndFlush(vresponse);
            		}
                break;
            case VOTE_RESPONSE:
            		EOrgVoteResponse vresponsed = (EOrgVoteResponse)msg;
            		if (logger.isDebugEnabled())
            			logger.debug("received vote response: {}", vresponsed.toString());
        			
            		if (vresponsed.getTickets() == 1 && myTickets.get() == 1) {
        				electionNode.state = Status.LOOKING;
            		} else if (vresponsed.getTickets() < myTickets.get()) {
        				myTickets.incrementAndGet();
        				electionNode.state = Status.SELECTED;
            		} 
                break;
            default:
                break;
        }
	}
	
	public boolean sendVoteRequest(EOrgVoteRequest request) {
		boolean flag = false;
		for (Channel channel : channelManager.getActivePeers()) {
			channel.getChannelPipeline().channel().writeAndFlush(request);
			//logger.info("send vote ticket to {}", channel.toString());
			flag = true;
        }
		return flag;
	}
	
	public Channel getActivePeer(Node node) {
		if (node == null) {
			return null;
		}
		Channel activePeer = channelManager.getActivePeer(node.getId());
		if (activePeer == null || activePeer.isDisconnected()) {
			channelManager.connect(node);
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			}
			activePeer = channelManager.getActivePeer(node.getId());
		}
		return activePeer;
	}
	
	public Channel getElectionPeer(byte[] nodeId) {
		return getActivePeer(getActivePeerNode(nodeId));
	}
	
	public Node getActivePeerNode(byte[] nodeId) {
		return channelManager.getNodeById(nodeId);
	}
	
	public boolean isActivePeer(Node node) {
		Channel activePeer = channelManager.getActivePeer(node.getId());
		if (activePeer == null || activePeer.isDisconnected()) {
			return false;
		}
		return true;
	}
	
}
